package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import com.atguigu.app.func.AsyncJoinFunction;
import com.atguigu.bean.TradeProvinceOrderWindow;
import com.atguigu.util.ClickHouseUtil;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class DwsTradeProvinceOrderWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        // TODO 3 读取topic_db数据
        String topicName = "topic_db";
        String groupId = "dws_trade_province_order_window";
        DataStreamSource<String> topicDbStream = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicName, groupId));


        SingleOutputStreamOperator<JSONObject> jsonObjStream = topicDbStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String table = jsonObject.getString("table");
                String type = jsonObject.getString("type");
                if ("order_info".equals(table) && "insert".equals(type)) {
                    out.collect(jsonObject);
                }
            }
        });
        //jsonObjStream.print("jsonObjStream");
        SingleOutputStreamOperator<TradeProvinceOrderWindow> flatMapStream = jsonObjStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getJSONObject("data").getString("user_id");
            }
        }).flatMap(new RichFlatMapFunction<JSONObject, TradeProvinceOrderWindow>() {
            ValueState<String> lastOrderDtState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> lastOrderDtDesc = new ValueStateDescriptor<>("last_order_dt", String.class);
                lastOrderDtDesc.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.days(1L)).build());
                lastOrderDtState = getRuntimeContext().getState(lastOrderDtDesc);
            }

            @Override
            public void flatMap(JSONObject value, Collector<TradeProvinceOrderWindow> out) throws Exception {
                String lastOrderDt = lastOrderDtState.value();
                Long tsLong = value.getLong("ts");
                String toDate = DateFormatUtil.toDate(tsLong);
                Long orderUserCount = 0L;
                JSONObject data = value.getJSONObject("data");
                if (lastOrderDt == null || !lastOrderDt.equals(toDate)) {
                    orderUserCount += 1;
                    lastOrderDtState.update(toDate);

                }
                out.collect(TradeProvinceOrderWindow.builder()
                        .orderCount(1L)
                        .orderUserCount(orderUserCount)
                        .orderAmount(data.getDouble("final_amount"))
                        .provinceId(data.getString("province_id"))
                        .ts(value.getLong("ts") * 1000)
                        .build());
            }
        });
        //flatMapStream.print("flatMapStream");
        // TODO 5 按照省份ID开窗聚合
        SingleOutputStreamOperator<TradeProvinceOrderWindow> reduceBeanStream = flatMapStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<TradeProvinceOrderWindow>() {
                    @Override
                    public long extractTimestamp(TradeProvinceOrderWindow element, long recordTimestamp) {
                        return element.getTs();
                    }
                })).keyBy(TradeProvinceOrderWindow::getProvinceId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<TradeProvinceOrderWindow>() {
                    @Override
                    public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow value1, TradeProvinceOrderWindow value2) throws Exception {
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        value1.setOrderUserCount(value1.getOrderUserCount() + value2.getOrderUserCount());
                        value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                        return value1;
                    }
                }, new WindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeProvinceOrderWindow> input, Collector<TradeProvinceOrderWindow> out) throws Exception {
                        TradeProvinceOrderWindow next = input.iterator().next();
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setTs(System.currentTimeMillis());
                        out.collect(next);
                    }
                });

        // reduceBeanStream.print("reduceBeanStream>");
        // TODO 6 维度关联省份名称
        SingleOutputStreamOperator<TradeProvinceOrderWindow> provinceBeanStream = AsyncDataStream.unorderedWait(reduceBeanStream, new AsyncJoinFunction<TradeProvinceOrderWindow>("dim_base_province".toUpperCase()) {
            @Override
            public String getKey(TradeProvinceOrderWindow input) {
                return input.getProvinceId();
            }

            @Override
            public void join(TradeProvinceOrderWindow input, JSONObject dimSkuInfo) {
                input.setProvinceName(dimSkuInfo.getString("name"));
            }
        }, 60L, TimeUnit.SECONDS);

        //provinceBeanStream.print("province>>>");

        // TODO 7 写出到clickHouse中
        provinceBeanStream.addSink(ClickHouseUtil.getClickHouseSink("insert into dws_trade_province_order_window values(?,?,?,?,?,?,?,?)"));

        // TODO 8 执行任务
        env.execute(groupId);

    }
}
